package maxent.mr.util;


import java.io.IOException;
import java.util.Map;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

/**
 *
 * @author hadoop
 */
public class MapSortHDFS {

    public static void MergeAndSort(String inputDir, String outputDir) throws IOException{

        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        FileStatus[] inputFiles = fs.listStatus(new Path(inputDir));
        Map<Integer, String> map=new TreeMap<Integer, String>();

        for(int i=0; i<inputFiles.length;i++){

            if(!inputFiles[i].getPath().getName().equalsIgnoreCase("_logs")){
                FSDataInputStream in = fs.open(inputFiles[i].getPath());
                String lines;

                while((lines=in.readLine()) != null) {
                    try{
                        map.put(getField(lines), getValue(lines));
                    }catch(Exception e){
                        e.getMessage();
                    }
                }
            }
        }
        
        FSDataOutputStream out = fs.create(new Path(outputDir));
        for(String val : map.values()){
                out.writeBytes(val);
                out.write('\n');
        }
    }

    private static int getField(String line) {
        return Integer.parseInt(line.split("\t")[0]);//extract value you want to sort on
    }

    private static String getValue(String line){
        return line.split("\t")[1];
    }
}
